package com.zusmart.base.network.nio;

import java.io.IOException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import com.zusmart.base.buffer.BufferAllocator;
import com.zusmart.base.logging.Logger;
import com.zusmart.base.logging.LoggerFactory;
import com.zusmart.base.looper.support.AbstractThreadEventLoop;
import com.zusmart.base.network.ChannelContext;
import com.zusmart.base.network.ChannelContextFuture;
import com.zusmart.base.network.ChannelOption;
import com.zusmart.base.network.ChannelUtils;

public class NioChannelEventLoop extends AbstractThreadEventLoop {

	private static final Logger logger = LoggerFactory.getLogger(NioChannelEventLoop.class);

	private Selector selector;
	private AtomicBoolean selecting = new AtomicBoolean(false);
	private Queue<ChannelContextFuture> doRegisterQueue = new ConcurrentLinkedQueue<ChannelContextFuture>();
	private Queue<ChannelContextFuture> deRegisterQueue = new ConcurrentLinkedQueue<ChannelContextFuture>();
	private Queue<NioChannelContextListenerFuture> doListenerQueue = new ConcurrentLinkedQueue<NioChannelContextListenerFuture>();

	public NioChannelEventLoop(String eventLoopName, NioChannelEventLoopGroup eventLoopGroup) {
		super(eventLoopName, eventLoopGroup);
	}

	public ChannelContextFuture doRegister(ChannelContext channelContext) {
		ChannelContextFuture future = new ChannelContextFuture(channelContext);
		if (this.inEventLoop()) {
			this.handleDoRegisterNow(future);
		} else {
			this.doRegisterQueue.add(future);
			this.doWakeupSelector();
		}
		return future;
	}

	public ChannelContextFuture deRegister(ChannelContext channelContext) {
		ChannelContextFuture future = new ChannelContextFuture(channelContext);
		if (this.inEventLoop()) {
			this.handleDeRegisterNow(future);
		} else {
			this.deRegisterQueue.add(future);
		}
		return future;
	}

	public NioChannelContextListenerFuture doListener(ChannelContext channelContext, int events) {
		NioChannelContextListenerFuture future = new NioChannelContextListenerFuture(channelContext, events);
		if (this.inEventLoop()) {
			this.handleDoListenerNow(future);
		} else {
			this.doListenerQueue.add(future);
		}
		return future;
	}

	public ChannelOption getChannelOption() {
		return this.getEventLoopGroup().getChannelOption();
	}

	public BufferAllocator getBufferAllocator() {
		return this.getEventLoopGroup().getBufferAllocator();
	}

	@Override
	protected void doStart() throws Exception {
		this.selector = Selector.open();
		super.doStart();
	}

	@Override
	protected void doClose() throws Exception {
		super.doClose();
	}

	@Override
	protected void doWake() {
		this.doWakeupSelector();
	}

	@Override
	protected void doLoop() throws Exception {

		this.handleProcessEvents();

		int selected = 0;
		try {
			if (this.selecting.compareAndSet(false, true)) {
				selected = this.selector.selectNow();
			} else {
				selected = this.selector.select(2000);
				this.selecting.set(false);
			}
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}

		if (selected > 0) {
			Set<SelectionKey> keys = this.selector.selectedKeys();
			for (SelectionKey key : keys) {
				ChannelContext channelContext = (ChannelContext) key.attachment();
				if (null == channelContext) {
					ChannelUtils.cancel(key);
					continue;
				}
				if (key.isValid() && key.isReadable()) {
					try {
						channelContext.doReadable();
					} catch (IOException e) {
						channelContext.doException(e);
					}
				} else if (key.isValid() && key.isWritable()) {
					try {
						channelContext.doWritable();
					} catch (IOException e) {
						channelContext.doException(e);
					}
				} else {
					ChannelUtils.cancel(key);
					channelContext.deRegister();
				}
			}
			keys.clear();
		}
	}

	@Override
	protected void doStop() throws Exception {
		ChannelUtils.close(this.selector);
	}

	@Override
	public NioChannelEventLoopGroup getEventLoopGroup() {
		return (NioChannelEventLoopGroup) super.getEventLoopGroup();
	}

	protected void handleProcessEvents() {
		ChannelContextFuture channelContextFuture;
		NioChannelContextListenerFuture nioChannelContextListenerFuture;
		while ((channelContextFuture = this.doRegisterQueue.poll()) != null) {
			this.handleDoRegisterNow(channelContextFuture);
		}
		while ((channelContextFuture = this.deRegisterQueue.poll()) != null) {
			this.handleDeRegisterNow(channelContextFuture);
		}
		while ((nioChannelContextListenerFuture = this.doListenerQueue.poll()) != null) {
			this.handleDoListenerNow(nioChannelContextListenerFuture);
		}
	}

	protected void handleDoRegisterNow(ChannelContextFuture channelContextFuture) {
		try {
			channelContextFuture.getChannelContext().doRegister(this.selector);
			channelContextFuture.setSuccess();
		} catch (IOException e) {
			channelContextFuture.getChannelContext().doException(e);
			channelContextFuture.setFailure(e);
		}
	}

	protected void handleDeRegisterNow(ChannelContextFuture channelContextFuture) {
		channelContextFuture.getChannelContext().deRegister();
		channelContextFuture.setSuccess();
	}

	protected void handleDoListenerNow(NioChannelContextListenerFuture channelContextFuture) {
		try {
			channelContextFuture.getChannelContext().doListener(channelContextFuture.getEvents());
			channelContextFuture.setSuccess();
		} catch (CancelledKeyException e) {
			channelContextFuture.getChannelContext().deRegister();
			channelContextFuture.setFailure(e);
		} catch (Throwable e) {
			channelContextFuture.getChannelContext().doException(e);
			channelContextFuture.setFailure(e);
		}
	}

	protected void doWakeupSelector() {
		this.selector.wakeup();
	}

	protected class ChannelUpdater {
		protected int events;
		protected ChannelContext channelContext;

		public ChannelUpdater(int events, ChannelContext channelContext) {
			this.events = events;
			this.channelContext = channelContext;
		}
	}

}